package com.ai.smlcore.service.common.impl;

import java.util.HashMap;
import java.util.Map;

import com.ai.frame.bean.InputObject;
import com.ai.frame.logger.Logger;
import com.ai.frame.logger.LoggerFactory;
import com.ai.frame.util.JsonUtil;
import com.ai.smlcore.service.common.IQueueService;
import com.ai.smlcore.util.PropertiesUtil;
import com.ai.smlcore.util.StringUtil;
import com.ai.smlcore.util.Constants.SYS_CONFIG;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.client.producer.SendStatus;
import com.alibaba.rocketmq.common.message.Message;

public class QueueServiceImpl implements IQueueService {
	private static final Logger logger = LoggerFactory.getServiceLog(QueueServiceImpl.class);
	private static Map<String, DefaultMQProducer> producerMap = new HashMap<String, DefaultMQProducer>();
	/**
	 * 获取生产者
	 * @param topic 主题
	 * @return 生产者
	 */
	private DefaultMQProducer getProducer (String topic) throws Exception {
		if (producerMap.containsKey(topic)) {
			return producerMap.get(topic);
		}
		DefaultMQProducer producer = new DefaultMQProducer(topic+"_PROD_GROUP");
		producer.setNamesrvAddr(PropertiesUtil.getString(SYS_CONFIG.REMOTE_MQ_IP_PORT));
		producer.setInstanceName(topic);
		producer.start();
		producerMap.put(topic, producer);
		return producer;
	}

	/**
	 * 将内容放入消息队列
	 * @param topic 主题
	 * @param key 业务标记
	 * @param inputObject 内容
	 * @return 
	 * @throws Exception
	 */
	public boolean sendMsg2MQ(String topic, String key, InputObject inputObject) throws Exception {
		if (StringUtil.isEmpty(topic) || inputObject == null) {
			throw new Exception("放入队列时,入参为空!");
		}
		DefaultMQProducer producer = getProducer(topic);
		Message msg = new Message(topic,null,key,JsonUtil.inputObject2Json(inputObject).getBytes("utf-8"));
		SendResult sendResult = producer.send(msg);
		logger.info("sendMsg2MQ参数", "topic="+topic+"\n  key="+key+" \n inputObject="+inputObject);
		logger.info("SendStatus", sendResult.getSendStatus().toString());
		return SendStatus.SEND_OK.equals(sendResult.getSendStatus());
	}
	
	public boolean sendMsg2MQ(String topic, String key, String value)
			throws Exception {
		if (StringUtil.isEmpty(topic) || StringUtil.isEmpty(value)) {
			throw new Exception("放入队列时,入参为空!");
		}
		DefaultMQProducer producer = getProducer(topic);
		Message msg = new Message(topic,null,key,value.getBytes("utf-8"));
		SendResult sendResult = producer.send(msg);
		logger.info("sendMsg2MQ参数", "topic="+topic+"\n  key="+key+" \n value="+value);
		logger.info("SendStatus", sendResult.getSendStatus().toString());
		return SendStatus.SEND_OK.equals(sendResult.getSendStatus());
	}
	
	/**
	 * 将内容放入消息队列
	 * @param topic 主题
	 * @param key 业务标记
	 * @param inputObject 内容
	 * @return 
	 * @throws Exception
	 */
	public boolean saveOrderToRedisMq(String topic, String key, InputObject inputObject) throws Exception {
		if (StringUtil.isEmpty(topic) || inputObject == null) {
			throw new Exception("放入队列时,入参为空!");
		}
		DefaultMQProducer producer = getProducer(topic);
		Message msg = new Message(topic, null,key,JsonUtil.inputObject2Json(inputObject).getBytes("utf-8"));
		SendResult sendResult = producer.send(msg);
		logger.info("sendMsg2MQ参数", "topic="+topic+"\n  key="+key+" \n inputObject="+inputObject);
		logger.info("SendStatus", sendResult.getSendStatus().toString());
		return SendStatus.SEND_OK.equals(sendResult.getSendStatus());
	}
}
